package io.kiki.sba.registry.server.clientInterface.node.service;

import io.kiki.sba.registry.api.Server;
import io.kiki.sba.registry.api.exchange.DataExchanger;
import io.kiki.sba.registry.common.model.Node;
import io.kiki.sba.registry.common.model.Node.NodeType;
import io.kiki.sba.registry.common.model.metaserver.inter.heartbeat.HeartBeatResponse;
import io.kiki.sba.registry.common.model.metaserver.inter.heartbeat.HeartbeatRequest;
import io.kiki.sba.registry.common.model.metaserver.nodes.ClientInterfaceNode;
import io.kiki.sba.registry.common.model.slot.SlotConfig;
import io.kiki.sba.registry.common.model.slot.Slots;
import io.kiki.sba.registry.common.model.store.Url;
import io.kiki.sba.registry.server.clientInterface.bootstrap.Config;
import io.kiki.sba.registry.server.clientInterface.multi.cluster.DataCenterCache;
import io.kiki.sba.registry.server.clientInterface.remoting.DataNodeExchanger;
import io.kiki.sba.registry.server.clientInterface.remoting.DataNodeNotifyExchanger;
import io.kiki.sba.registry.server.clientInterface.slot.SlotsCache;
import io.kiki.sba.registry.server.shared.env.ServerEnv;
import io.kiki.sba.registry.server.shared.meta.AbstractMetaServerService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.Set;


public class MetaServerServiceImpl extends AbstractMetaServerService<HeartBeatResponse> {
    protected final Logger logger = LoggerFactory.getLogger(AbstractMetaServerService.class);

    @Autowired
    private Config config;

    @Autowired
    private SlotsCache slotsCache;

    @Autowired
    private DataCenterCache dataCenterCache;

    @Autowired
    private DataNodeExchanger dataNodeExchanger;

    @Autowired
    private DataNodeNotifyExchanger dataNodeNotifyExchanger;

    @Autowired
    private DataExchanger boltDataExchanger;

    @Override
    protected long getCurrentslotsEpoch() {
        return slotsCache.getEpoch(config.getDataCenter());
    }

    @Override
    public int getRenewIntervalSecs() {
        return config.getSchedulerHeartbeatIntervalSecs();
    }

    @Override
    protected void handleRenewResult(HeartBeatResponse heartBeatResponse) {
        Set<String> dataServerList = getDataServerList();
        if (dataServerList != null && !dataServerList.isEmpty()) {
            dataNodeNotifyExchanger.setServerIpSet(dataServerList);
            dataNodeNotifyExchanger.notifyConnectServerAsync();
            dataNodeExchanger.setServerIpSet(dataServerList);
            dataNodeExchanger.notifyConnectServerAsync();
        }
        if (heartBeatResponse.getSlots() != null && heartBeatResponse.getSlots() != Slots.INIT) {
            slotsCache.updateLocalSlots(heartBeatResponse.getSlots());
        } else {
            logger.warn("[handleRenewResult] no slot table result");
        }

        slotsCache.updateRemoteSlots(heartBeatResponse.getRemoteSlotsStatusMap());
        dataCenterCache.saveDataCenterList(heartBeatResponse.getRemoteSlotsStatusMap());
    }

    @Override
    protected HeartbeatRequest createRequest() {
        return new HeartbeatRequest(createNode(), slotsCache.getEpoch(config.getDataCenter()), config.getDataCenter(), System.currentTimeMillis(), SlotConfig.slotBasicInfo(), slotsCache.getRemoteslotsEpoch()).setSlots(slotsCache.getLocalSlots());
    }

    @Override
    protected NodeType nodeType() {
        return NodeType.client_interface;
    }

    @Override
    protected String cell() {
        return commonConfig.getRegion();
    }

    private Node createNode() {
        return new ClientInterfaceNode(new Url(ServerEnv.IP), config.getSessionServerRegion(), ServerEnv.PROCESS_ID, getWeight());
    }

    private int getWeight() {
        Server server = boltDataExchanger.getServer(config.getServerPort());
        if (null == server) {
            return 0;
        }
        return server.getChannelList().size();
    }


}
